//package com.open.source.demo;
//
//import com.open.source.MqConstant;
//import com.open.source.MqManagerEnum;
//import com.open.source.kafka.AbstractKafkaDltConsumerListener;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.producer.ProducerRecord;
//import org.apache.kafka.common.header.Headers;
//import org.apache.kafka.common.header.internals.RecordHeader;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.stereotype.Component;
//
///**
// * 手动提交
// *
// * @author ZonLen since on 2022/7/29 下午12:22
// */
//
//@Slf4j
//@Component
//public class DemoKafkaDltConsumerListener extends AbstractKafkaDltConsumerListener {
//
//
//  @Override
//  public String[] topics() {
//    return new String[]{MqManagerEnum.TAX_TOPIC_DELAY_DEMO.getTopic()};
//  }
//
//  @Override
//  public void onMessage(ConsumerRecord<String, String> record) {
//    log.info("header host-ip={}", getHeaderByKey("host-ip", record));
//    final String timeStr = getHeaderByKey(MqConstant.X_RETRIES_HEADER, record);
//    log.info("header time={}", timeStr);
//    log.info("Manual offset ＝{}, key ＝{}, value ＝{}", record.offset(), record.key(),
//        record.value());
//    record.headers().add(new RecordHeader(MqConstant.X_RETRIES_HEADER,
//        String.valueOf(Integer.parseInt(timeStr) + 1).getBytes()));
//    int a = 1 / 0;
//  }
//
//  @Override
//  public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
//    final String timeStr = getHeaderByKey(MqConstant.X_RETRIES_HEADER, record);
//    final Headers headers = record.headers();
//    headers.add(new RecordHeader(MqConstant.X_RETRIES_HEADER, String.valueOf(Integer.parseInt(timeStr)+1).getBytes()));
//    new ProducerRecord<>(record.topic(), record.partition(),System.currentTimeMillis(), record.key(), record.value(), headers);
//    applicationContext.getBean(KafkaTemplate.class).send(record.topic()+".DLT", headers);
//  }
//}
